package io.wen.bd.s7m1

import org.apache.spark.sql.streaming.Trigger
import org.apache.spark.sql.{Dataset, SparkSession}

object StuctureStreamingMysql {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .appName("Spark Foreach MySql")
      .master("local[*]")
      .getOrCreate()

    spark.sparkContext.setLogLevel("WARN")

    import spark.implicits._

    val kafkaDF = spark.readStream
      .format("kafka")
      .option("kafka.bootstrap.servers", "hadoop03:9092")
      .option("subscribe", "test")
      .load()

    val kafkaValDF = kafkaDF.selectExpr("CAST(value AS STRING)")
    val kafkaDS = kafkaValDF.as[String]
    val busInfoDS = kafkaDS.map(msg => BusInfo(msg))

    val writer = new MysqlWriter

//    kafkaValDF.writeStream
//      .format("console")
//      .trigger(Trigger.ProcessingTime("2 seconds"))
//      .start()
//      .awaitTermination()

    busInfoDS.writeStream
      .foreach(writer)
      .outputMode("append")
      .start()
      .awaitTermination()
  }
}
